package com.ziroom.crm.mq.rabbit.spring.core;

import com.ziroom.crm.mq.core.resend.ReSendInvoker;
import com.ziroom.crm.mq.core.support.ProcessorInvokerDelegate;
import com.ziroom.crm.mq.rabbit.spring.data.MqRabbitConst;
import com.ziroom.crm.mq.rabbit.spring.data.RabbitMessage;
import com.ziroom.crm.mq.rabbit.spring.resolve.RabbitResolve;
import com.ziroom.crm.mq.rabbit.spring.wrapper.RabbitSendQueueWrapper;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;

import java.util.List;
import java.util.Objects;
import java.util.function.Consumer;

/**
 * 模板对象
 *
 * @author niXueChao
 * @date 2019/7/10
 */
@Slf4j
public class ReliableRabbitTemplate extends RabbitTemplate implements InitializingBean {

    @Autowired
    private ProcessorInvokerDelegate facade;

    /**
     * 回调方法
     */
    @Getter
    @Setter
    public Consumer<List<? extends com.ziroom.crm.mq.core.data.Message>> callback;

    /**
     * 数量回调
     */
    @Getter
    @Setter
    public Consumer<Integer> countCallback;

    /**
     * 初始化完成后执行
     */
    @Override
    public void afterPropertiesSet() {
        if (Objects.nonNull(callback)) {
            ReSendInvoker.setCallback(callback);
        }
        if (Objects.nonNull(countCallback)) {
            ReSendInvoker.setCountCallback(countCallback);
            //log.debug("设置countCallback：{}", countCallback);
        }
        super.afterPropertiesSet();
    }


    // 重写
    @Override
    public void send(String exchange, String routingKey, Message message, final CorrelationData correlationData) throws AmqpException {
        //生成自定义消息体
        RabbitMessage rabbitMessage = RabbitResolve.convert(message);
        rabbitMessage.setExchange(exchange);
        rabbitMessage.setRoutingKey(routingKey);

        RabbitSendQueueWrapper rabbitQueueWrapper = RabbitResolve.buildSendQueueWrapper(rabbitMessage);
        // 设置回调
        rabbitQueueWrapper.setRetryCallback((retryContext) -> {
            //消息中设置内存重试次数
            message.getMessageProperties().setHeader(MqRabbitConst.MQ_SEND_RETRY_COUNT, retryContext.getRetryCount());
            super.send(exchange, routingKey, message, correlationData);
        });

        // 是否为重试消息
        boolean isRetry = RabbitResolve.isResendMessage(message);
        if (!isRetry) {
            facade.preSendInvoke(rabbitQueueWrapper, rabbitMessage);
        }
        try {
            super.send(exchange, routingKey, message, correlationData);
        } catch (Exception e) {
            rabbitMessage.setException(e);
            // 如果是持久化重试，抛出异常，交于上级方法处理
            if (isRetry) {
                throw new RuntimeException(e.getMessage(), e.getCause());
            }
        } finally {
            //执行后置处理
            if (!isRetry) {
                facade.postSendInvoke(rabbitQueueWrapper, rabbitMessage);
            }
        }
    }

}
